package cn.hyxy.hadoop;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Demo06_ResortMR extends Configured implements Tool{

	public static void main(String[] args) throws Exception{
		int code=ToolRunner.run(new Demo06_ResortMR(), args);
		System.exit(code);
	}

	@Override
	public int run(String[] args) throws Exception {
		if (args.length!=2) {
			System.out.println("Usage...");
			return -1;
		}
		
		//获取配置信息
		Configuration configuration=new Configuration();
		
		//如果输入路径存在文件，先删除
		FileSystem fSystem=FileSystem.get(configuration);		
		Path path=new Path(args[1]);
		if (fSystem.exists(path)) {
			fSystem.delete(path, true); //true：递归文件--子文件
		}
		
//		Job job=new Job(configuration, "ResortMR");
		Job job = Job.getInstance(configuration, "ResortMR");		
		job.setJarByClass(getClass());
		
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(LongWritable.class);
		
		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, path);
		
		Boolean flag=job.waitForCompletion(true);
		
		return flag?0:1;
	}
	
	public static class MyMapper extends Mapper<LongWritable , Text, Text, LongWritable>{
		private Text key2=new Text();
		private LongWritable value2=new LongWritable(1);
		private String fileName=null;
		@Override
		protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			InputSplit inputSplit=context.getInputSplit();
			FileSplit fSplit=(FileSplit) inputSplit;
			fileName=fSplit.getPath().getName();//获取文件名；	
			System.out.println("fileName:"+fileName);
		}
		
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			String[] strings=value.toString().split("\\s+");
			for (String string : strings) {
				if (string.trim().equals("")) {  //去掉空格
					continue;
				}
				key2.set(fileName+"\t"+string);
				context.write(key2, value2);
			}
		}
	}
	
	public static class MyReducer extends Reducer< Text, LongWritable, Text, NullWritable>{
		
		/**
		 * 法二）；输出的是：Text  Null
		 */
	 	private Text key4 = new Text();

	        @Override
	        protected void reduce(Text key3, Iterable<LongWritable> value3, Context context) throws IOException, InterruptedException {
	            long sum = 0;
	            for (LongWritable w : value3) {
	                sum += w.get();
	            }
	            key4.set(key3.toString() + "\t" + sum);//1.txt Jack 2
	            context.write(key4, NullWritable.get());
	        }
	        /*	@Override
		protected void reduce(Text key3, Iterable<LongWritable> value3,
				Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
			long sum=0;
			for (LongWritable longWritable : value3) {
				sum+=longWritable.get();
			}
			context.write(key3, new LongWritable(sum));
		}*/
	}

}
